package com.ruirui.demo;

import reactor.core.publisher.Flux;

public class FluxLimitRate {
    public static void main(String[] args) throws InterruptedException {
        /**
         * 确保下游订阅者发出的背压信号在向上传播时被分割成不超过所提供的 prefetchRate 的批次，从而有效地对上游 Publisher 进行速率限制。
         * 请注意，这是一个上限，该操作符使用预取-补充策略，在已发射预取量的75%时请求补充量。
         * 通常用于消费者请求大量数据（例如 Long.MAX_VALUE）但数据源在较小请求时表现更好或可以优化的场景（例如数据库分页等）。
         * 所有数据仍会被处理，这与 take(long) 不同，take(long) 会限制总的请求量。
         * 等效于 flux.publishOn(Schedulers.immediate(), prefetchRate).subscribe()。请注意，prefetchRate 是一个上限，此操作符使用预取-补充策略，
         * 在已发射预取量的75%时请求补充量。
         */

        /**
         * limitRate 是一个上游限速操作符：
         * - 将下游（Subscriber）请求的信号切分为多个批次（不超过 prefetchRate）；
         * - 每批最多请求 `prefetchRate` 个元素；
         * - 默认在已消费当前批次的 75% 时请求下一批（即默认补充阈值是 75%）。
         */
        Flux<Integer> flux = Flux.range(1, 1000);
       flux // 创建一个发射 1 到 1000 共 1000 个整数的 Flux 流
                .log()                   // 打印所有订阅、请求、发射、取消等 Reactive Signals（方便调试）
                .limitRate(100) // 设置背压速率上限，每次最多向上游请求 100 个元素（默认 75 个消费后补充）
                .subscribe();            // 订阅该流，开始触发数据流动

        /**
         * | 操作符         | 作用                    | 会不会丢弃多余数据 | 举例说明                    |
         * | ----------- | --------------------- | --------- | ----------------------- |
         * | `limitRate(100)` | **分批次请求数据**，全部数据仍然处理  | ❌ 不会丢数据   | 请求 100，消费 75 就触发下一个 100 |
         * | `take(100)` | **只处理前 N 个数据，之后取消订阅** | ✅ 会丢弃剩余   | 只消费前 100 个，其余不会被处理      |
         */
        flux
               .log()
               .take(100)
               .subscribe();

    }
}
